package mapred.joinmode;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;

public class UserJoinReducer extends Reducer<Text, Text, Text, Text> {
    private static final Text EMPTY_TEXT = new Text("\t\t\t\t\t\t");
    private Text tmp = new Text();
    private ArrayList<Text> listA = new ArrayList<Text>();
    private ArrayList<Text> listB = new ArrayList<Text>();
    private String joinType = null;

    public void setup(Context context) {
        // Get the type of join from our configuration
        joinType = context.getConfiguration().get("join.type");
    }

    //我们以Key=1为例，分析数据处理的过程：
    // 当前reduce函数接收到的数据是：
    //1：A1	张三	34	1990-1-1
    //    B1	2018级	大数据班
    //2：A2	李四	23	2000-1-1
    //    B2	2017级	Java班
    //3：A3	王五	12	2010-3-3
    //    B3	2017级	微软班
    //4：A4	赵六	33  0000-0-0
    //5：B5	2016级	未知
    // 也就是说：每个Key的value数组中，最多由2条数据
    public void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {
        // Clear our lists
        // 在处理当前key的数据时，先将全局的临时变量清空
        // 按照算法来分析，listA 和listB最多由1条数据
        // 之所以用数组来存数据，就是为了模拟连接中的1：n的模式
        listA.clear();
        listB.clear();
        // iterate through all our values, binning each record based on what
        // it was tagged with. Make sure to remove the tag!

        // 循环Key的value值，将A和B根据前缀不同，分配到不同的数组中
        while (values.iterator().hasNext()) {
            tmp = values.iterator().next();
            // 根据前缀的不同，将数据放到listA 和listB中
            // 原来是2个Mapper处理2个文件中的数据，到这里，才开始为建立2个数据的连接做准备
            if (tmp.charAt(0) == 'A') {
                listA.add(new Text(tmp.toString().substring(1)));
            } else if (tmp.charAt(0) == 'B') {

                // 当key=4的时候，listA 会放入一个值，但是listB没有值
                // 当key=5的时候，listB会放入一个值
                listB.add(new Text(tmp.toString().substring(1)));
            }
        }
        // Execute our join logic now that the lists are filled
        executeJoinLogic(context);
    }

    private void executeJoinLogic(Context context) throws IOException, InterruptedException {


        // joinType是在setup函数中，从配置中获取到的连接类型

        // 如果是内连接:
        // A和B必须都有相同的数据，参会将此条数据放入结果中
        if (joinType.equalsIgnoreCase("inner")) {
            // If both lists are not empty, join A with B
            if (!listA.isEmpty() && !listB.isEmpty()) {
                for (Text A : listA) {
                    for (Text B : listB) {

                        // 双重循环，实现了将A中的数据和B中的数据实现笛卡尔积
                        // 输出的数据条数为：A * B的数量
                        // 输出的数据特征是：
                        // 1	张三	34	1990-1-1    1	2018级	大数据班
                        context.write(A, B);
                    }
                }
            }
        }
        // Left outer connection, right list empty, output empty value：
        // 如果是左连接:
        // A必须输出到结果中，B如果没有值，直接输出空值
        else if (joinType.equalsIgnoreCase("leftouter")) {
            // For each entry in A,
            // 循环遍历listA,保证A中的数据一定能输出
            for (Text A : listA) {
                // If list B is not empty, join A and B
                // 如果B中也有数据，那么将A+B一起输出到结果中
                if (!listB.isEmpty()) {
                    for (Text B : listB) {
                        context.write(A, B);
                    }
                } else {
                    // Else, output A by itself
                    // 如果B中没有数据，那么只A输出到结果中，B输出为空值
                    // 数据的输出特征是：
                    // 4	赵六	33  0000-0-0    （右侧是空值）
                    context.write(A, EMPTY_TEXT);
                }
            }
        }
        // Right External Connection：
        // 如果是左连接:
        // B必须输出到结果中，A如果没有值，直接输出空值
        else if (joinType.equalsIgnoreCase("rightouter")) {
            // For each entry in B,
            // 循环遍历listB,保证B中的数据一定能输出
            for (Text B : listB) {
                // If list A is not empty, join A and B
                // 如果A中也有数据，那么将A+B一起输出到结果中
                if (!listA.isEmpty()) {
                    for (Text A : listA) {
                        context.write(A, B);
                    }
                } else {
                    // Else, output B by itself
                    // 输出的值的特征是：
                    //                  5	2016级	未知
                    context.write(EMPTY_TEXT, B);
                }
            }
        }

    }
}